import com.zuikaku.pojo.Order;
import com.zuikaku.pojo.People;
import com.zuikaku.sink.OrderSink;
import com.zuikaku.source.OrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import javax.xml.crypto.Data;
import java.util.Arrays;

public class SourceAPIDemo {

    public static void main(String[] args){
        //Flink流处理案例
        //1.取得流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);
//        //2.获得source
//        DataStream<String> source1 = env.fromElements("zhangsan,18,100","lisi,20,120","wangwu,15,60");
//        source1.print("源数据(source)1");
//        DataStream<String> source2 = env.fromCollection(Arrays.asList("zhangsan,18,100","lisi,20,120","wangwu,15,60"));
//        source2.print("源数据(source)2");
//        DataStream<Long> source3 = env.fromSequence(1,5);
//        source3.print("源数据(source)3");
//        //来自本地文本文件
//        DataStream<String> source4 = env.readTextFile("C:\\Users\\Zuikaku\\Desktop\\test.txt");
//        source4.print("源数据(source)4");
        //来自于自定义流
        DataStream<Order> source5 = env.addSource(new OrderSource());
        source5.print("源数据(source)5");
        //source5.printToErr()
        DataStream<Order> iphoneOrderDS = source5.filter(new FilterFunction<Order>() {
            @Override
            public boolean filter(Order value) throws Exception {
                return value.getItemName().equals("iPhone12");
            }
        }).setParallelism(3);
        iphoneOrderDS.print("iphone订单source").setParallelism(4);
        //添加自定义Sink进行消费
        iphoneOrderDS.addSink(new OrderSink());

        try {
            env.execute("source api demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
